#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <fcntl.h>
#include <unistd.h>
#include <time.h>
#include <sys/epoll.h>
#include <signal.h>
#include <sys/signalfd.h>
#include <errno.h>

#include "logm.h"
#include "../src/conn/conn.h"
#include "../src/config/config.h"


typedef struct
{
    char    log_path[1024];
    char    sock_path[1024];
    long    mtime;
    FILE*   kmsg;
    int     kmsgfd;
    int     epoll_fd;
    int     pid;
}logm_t;



typedef struct{
    int  fd;
    char buffer[4096]; 
    char name[1024];
}evts_data;

logm_t logmor;


ssize_t readn(int fd, void *buf, size_t count){
    size_t nleft = count;
    ssize_t nread;
    char *bufp = (char*)buf;

    while(nleft > 0){
        if((nread = read(fd, bufp, nleft)) < 0){
            if(errno == EINTR)
                continue;
            return -1;
        } else if (nread == 0){
            return count - nleft;
        }
        bufp += nread;
        nleft -= nread;
    }
    return count;
}

//检查socket缓冲区
ssize_t recv_peek(int sockfd, void* buf, size_t len){
    while(1){
        int ret = recv(sockfd, buf, len, MSG_PEEK);
        if(ret == -1 && errno == EINTR)
            continue;
        return ret;
    }
}

//按行读取socket
ssize_t readline(int sockfd, void* buf, size_t maxline){
    int ret;
    int nread;
    char *bufp = buf;
    int nleft = maxline;
    while(1){
        ret = recv_peek(sockfd, bufp, nleft);
        if(ret < 0)
            return ret;
        else if(ret == 0){
            return ret;
        }
        nread = ret;
        int i;

        for(i = 0; i<nread; i++){
            if(bufp[i] == '\n'){
                ret = readn(sockfd, bufp, i+1);
                if(ret != i+1){
                    return -1;
                    //exit(EXIT_FAILURE);
                }
                bufp[i+1]='\0';
                return ret;
            }
        }

        if(nread > nleft)
            return -1;
            //exit(EXIT_FAILURE);
        nleft -= nread;
        ret = readn(sockfd, bufp, nread);
        if(ret != nread)
            return -1;
            //exit(EXIT_FAILURE);

        bufp += nread;
    }
    return -1;
}

//发送fd
ssize_t send_fd(int sock_fd, int send_fd,char strmsg[1024])
{
    int ret;
    struct msghdr msg;
    struct cmsghdr *p_cmsg;
    struct iovec vec;
    char cmsgbuf[CMSG_SPACE(sizeof(send_fd))];
    int *p_fds;
    msg.msg_control = cmsgbuf;
    msg.msg_controllen = sizeof(cmsgbuf);
    p_cmsg = CMSG_FIRSTHDR(&msg);
    p_cmsg->cmsg_level = SOL_SOCKET;
    p_cmsg->cmsg_type = SCM_RIGHTS;
    p_cmsg->cmsg_len = CMSG_LEN(sizeof(send_fd));
    p_fds = (int *)CMSG_DATA(p_cmsg);
    *p_fds = send_fd; // 通过传递辅助数据的方式传递文件描述符
    msg.msg_name = NULL;
    msg.msg_namelen = 0;
    
    msg.msg_iov = &vec;
    msg.msg_iovlen = 1; //主要目的不是传递数据，故只传1个
    msg.msg_flags = 0;
    char charbuf[1024];
    if (strmsg!=NULL)
    {
        strcpy(charbuf,strmsg);
        vec.iov_base = charbuf;
        vec.iov_len = strlen(strmsg)+1;   
    }else{
        char charbuf= 0;
        vec.iov_base = &charbuf;
        vec.iov_len = sizeof(charbuf); 
    }
    ret = sendmsg(sock_fd, &msg, 0);
    if (ret < 1)
        printf("sendmsg err\n");
    return ret;
}

//接收fd
int recv_fd(const int sock_fd,char *strmsg)
{
    int ret;
    struct msghdr msg;
    struct iovec vec;
    int recv_fd;
    char cmsgbuf[CMSG_SPACE(sizeof(recv_fd))];
    struct cmsghdr *p_cmsg=NULL;
    int *p_fd;
    if (strmsg!=NULL)
    {
        vec.iov_base = strmsg;
        vec.iov_len = 1024;  
    }else{
        char charbuf;
        vec.iov_base = &charbuf;
        vec.iov_len = 1;
    }
    
    msg.msg_name = NULL;
    msg.msg_namelen = 0;
    msg.msg_iov = &vec;
    msg.msg_iovlen = 1;
    msg.msg_control = cmsgbuf;
    msg.msg_controllen = sizeof(cmsgbuf);
    msg.msg_flags = 0;
    
    
    p_fd = (int *)CMSG_DATA(CMSG_FIRSTHDR(&msg));
    *p_fd = -1;
    ret = recvmsg(sock_fd, &msg, 0);
    if (ret > 0)
        if (strmsg!=NULL)
            //strmsg[ret]='\0';
            //printf("接data: %d %s\n",ret,strmsg);
        
        

    p_cmsg = CMSG_FIRSTHDR(&msg);
    if (p_cmsg == NULL)
        printf("no passed fd\n");


    p_fd = (int *)CMSG_DATA(p_cmsg);
    recv_fd = *p_fd;
    if (recv_fd == -1)
        printf("no passed fd\n");

    return recv_fd;
}

//epoll事件控制
static void epoll_add_event(int epollfd,int fd,void *data,int mode){
    struct epoll_event ev;
    ev.events = mode;
    ev.data.ptr = data;
    epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&ev);    
}


void sig_handler(int sig){
    unlink(logmor.sock_path);
    fprintf(logmor.kmsg,"logm[%d]: Exited Signal=%d\n",logmor.pid,sig);
    exit(128+sig);  
}

//解析logm配置
void parse_logmconf(void *conf,const char* section,const char* key,char* value){
    logm_t *logmor=(logm_t*)conf;
    if (strcmp(section,"logm")==0 && strcmp(key,"Socket")==0)
    {
        trim(value);
        strcpy(logmor->sock_path,value);
    }else if (strcmp(section,"logm")==0 && strcmp(key,"LogPath")==0)
    {
        trim(value);
        strcpy(logmor->log_path,value);
    }
}


int log_server() 
{
    //添加加载配置文件
    strcpy(logmor.sock_path,"/run/logm.sock");
    strcpy(logmor.log_path,"/tmp/message");
    logmor.pid=getpid();
    if (access("/root/rinitd/etc/rinitd/conf/logm.conf",F_OK)==0)
    {
        ini_parse(&logmor,"/root/rinitd/etc/rinitd/conf/logm.conf",parse_logmconf,&logmor.mtime);
    }
    logmor.kmsg=fopen("/dev/kmsg","a+");
    if (logmor.kmsg!=NULL)
    {
        fprintf(logmor.kmsg,"logm[%d]: Logm Server Started.\n",logmor.pid);
        fflush(logmor.kmsg);
    }

    // 创建 socket
    int server_socket, client_fd;
    struct sockaddr_un addr;
    if ((server_socket = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }

    // 绑定 socket
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, logmor.sock_path, sizeof(addr.sun_path) - 1);
    unlink(logmor.sock_path);

    if (bind(server_socket, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
        perror("bind failed");
        close(server_socket);
        exit(EXIT_FAILURE);
    }

    // 监听
    if (listen(server_socket, 50) < 0) {
        perror("listen");
        close(server_socket);
        exit(EXIT_FAILURE);
    }

    //创建epoll 启动unit并且设置子进程io
    struct epoll_event evts[20];
    logmor.epoll_fd=epoll_create(1);
    evts_data *data=malloc(sizeof(evts_data));
    data->fd=server_socket;
    strcpy(data->name,"");
    epoll_add_event(logmor.epoll_fd,server_socket,data,EPOLLIN|EPOLLET);

    // log持久化
    FILE *logpath=fopen(logmor.log_path,"a");
    signal(SIGTERM,sig_handler);
    signal(SIGINT,sig_handler);
    signal(SIGSEGV,sig_handler);
    while (1)
    {
       // printf("新建连接client hhhhhhhhh fd:%d\n",client_fd);
        int ret=epoll_wait(logmor.epoll_fd,evts,20,-1);
        //printf("epoll ret:%d\n",ret);
        for (int i = 0; i < ret; i++)
        {
            evts_data *epptr=(evts_data*)evts[i].data.ptr;
            if (epptr->fd==server_socket) //socket触发连接事件
            {
                //socket新建连接
                if ((client_fd = accept4(server_socket, NULL, NULL,SOCK_CLOEXEC|SOCK_NONBLOCK)) <= 0) {
                    if (logmor.kmsg!=NULL)
                    {
                        fprintf(logmor.kmsg,"logm[%d]: socket %s accept %d:\n",logmor.pid,strerror(errno),client_fd);
                        fflush(logmor.kmsg);
                    }
                }else{
                    int remote_fd;
                    char msg[1024];
                    if((remote_fd=recv_fd(client_fd,msg))>0){
                        evts_data *data=malloc(sizeof(evts_data));
                        data->fd=remote_fd;
                        strcpy(data->name,msg);
                        epoll_add_event(logmor.epoll_fd,remote_fd,data,EPOLLIN|EPOLLERR|EPOLLRDHUP);
                        if (logmor.kmsg!=NULL)
                        {
                            fprintf(logmor.kmsg,"logm[%d]: epoll add %s fd=%d\n",logmor.pid,data->name,remote_fd);
                            fflush(logmor.kmsg);
                        }
                    }
                }
                close(client_fd);
            }else if (epptr->fd!=server_socket&&(evts[i].events&EPOLLIN)) //remote_fd触发读事件
            {
                ssize_t ret=readline(epptr->fd,epptr->buffer,LOG_BUFFER_SIZE-1);
                if(ret>0)
                {
                    time_t now=time(NULL);
                    struct tm *local_time=localtime(&now);
                    fprintf(logpath,"<%d-%02d-%02d %02d:%02d:%02d %s>::   %s",local_time->tm_year+1900,local_time->tm_mon,local_time->tm_mday,local_time->tm_hour,local_time->tm_min,local_time->tm_sec,epptr->name,epptr->buffer);
                    //printf("%02d-%02d %02d:%02d:%02d Recv.%s: %s",local_time->tm_mon,local_time->tm_mday,local_time->tm_hour,local_time->tm_min,local_time->tm_sec,epptr->name,epptr->buffer);
                    fflush(logpath);
                }else{
                    //关闭socketpair fd并释放epoll ptr
                    epoll_ctl(logmor.epoll_fd,EPOLL_CTL_DEL,epptr->fd,NULL);
                    close(epptr->fd);
                    free(epptr);
                    if (logmor.kmsg!=NULL)
                    {
                        fprintf(logmor.kmsg,"logm[%d]: %s removed msg(%s)\n",logmor.pid,epptr->name,strerror(errno));
                        fflush(logmor.kmsg);
                    }  
                }
            }else if (epptr->fd!=server_socket&&((evts[i].events&EPOLLRDHUP) || (evts[i].events&EPOLLERR)))
            {
                    //关闭socketpair fd并释放epoll ptr
                    fprintf(logmor.kmsg,"epoll: %s removed fd:%d\n",strerror(errno),epptr->fd);
                    fflush(stdout);
                    epoll_ctl(logmor.epoll_fd,EPOLL_CTL_DEL,epptr->fd,NULL);
                    close(epptr->fd);
                    free(epptr);
            }else if (epptr->fd==server_socket && (evts[i].events&EPOLLHUP))
            {
                //logm 监听异常触发重建socket
                epoll_ctl(logmor.epoll_fd,EPOLL_CTL_DEL,epptr->fd,NULL);
                close(epptr->fd);
                free(epptr);
                if (logmor.kmsg!=NULL)
                {
                    fprintf(logmor.kmsg,"logm[%d]: msg(%s) removed %s fd=%d\n",logmor.pid,strerror(errno),epptr->name,epptr->fd);
                    fflush(logmor.kmsg);
                }  
                //logm 监听异常触发重建socket
                // unlink(logm_sock); 
                // if ((server_socket = socket(AF_UNIX, SOCK_STREAM, 0)) == 0) {
                //     perror("socket failed");
                //     continue;
                // }
                // if (bind(server_socket, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
                //     perror("bind failed");
                //     close(server_socket);
                //     continue;
                // }
                // if (listen(server_socket, 15) < 0) {
                //     perror("listen");
                //     close(server_socket);
                //     continue;
                // }
                // epoll_add_event(epoll_fd,server_socket,NULL,EPOLLIN|EPOLLET|EPOLLHUP);   
            }  
        }
    }
    // 关闭连接
    close(server_socket);
    fclose(logmor.kmsg);
    unlink(logmor.sock_path);
    return 0;
}


int handle_fd(char *sockpath,char *msg){
    int client_fd;
    struct sockaddr_un addr;

    // 创建 socket
    if ((client_fd = socket(AF_UNIX, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
            return -1;
    }

    // 连接服务器
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, sockpath, sizeof(addr.sun_path) - 1);

    if (connect(client_fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
        perror("connect failed");
        close(client_fd);
        return -1;
    }
    // 向服务器注册remote_fd
    int sock_peer[2];
    if (socketpair(AF_UNIX,SOCK_STREAM,0,sock_peer)==0)
    {
        // 客户注册remote_fd
        ssize_t ret=send_fd(client_fd,sock_peer[1],msg);
        if (ret>0)
        {
            close(sock_peer[1]);
            close(client_fd);
            return sock_peer[0];
        }else{
            close(sock_peer[1]);
            close(client_fd);
            printf("sendmsg: %s\n",strerror(errno));
            fflush(stdout);
            return -1;
        }
    }
    close(client_fd);
    return -1;
}


void ioworker(void* user,char *buf,FILE *out){
    epmsg_t *epmsg=(epmsg_t *)user;
    time_t now=time(NULL);
    struct tm *local_time=localtime(&now);
    fprintf(out,"<%d-%02d-%02d %02d:%02d:%02d %s>::   %s",local_time->tm_year+1900,local_time->tm_mon,local_time->tm_mday,local_time->tm_hour,local_time->tm_min,local_time->tm_sec,epmsg->name,buf);
    //printf("%02d-%02d %02d:%02d:%02d Recv.%s: %s",local_time->tm_mon,local_time->tm_mday,local_time->tm_hour,local_time->tm_min,local_time->tm_sec,epptr->name,epptr->buffer);
    fflush(out);
}

// logm v2
int logm_server() 
{
    // signal(SIGTERM,sig_handler);
    // signal(SIGINT,sig_handler);
    //添加加载配置文件
    strcpy(logmor.sock_path,"/run/logm.sock");
    strcpy(logmor.log_path,"/tmp/message");
    logmor.pid=getpid();
    if (access("/root/rinitd/etc/rinitd/conf/logm.conf",F_OK)==0)
    {
        ini_parse(&logmor,"/root/rinitd/etc/rinitd/conf/logm.conf",parse_logmconf,&logmor.mtime);
    }
    if ((logmor.kmsg=fopen("/dev/kmsg","a"))!=NULL)
    {
        fprintf(logmor.kmsg,"logm[%d]: Logm Server Started.\n",logmor.pid);
        fflush(logmor.kmsg);
    }
    //创建epoll 启动unit并且设置子进程io
    logmor.epoll_fd=epoll_create1(EPOLL_CLOEXEC);
    // 注册signalfd 
    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask,SIGINT);
    sigaddset(&mask,SIGQUIT);
    sigaddset(&mask,SIGTERM);
    if (sigprocmask(SIG_BLOCK, &mask, NULL) == -1){
        exit(EXIT_FAILURE);
    }
    int sfd=signalfd(-1, &mask, SFD_NONBLOCK|SFD_CLOEXEC);
    if (sfd<0)
    {
        exit(EXIT_FAILURE);
    }
    epmsg_t *epmsg=epmsg_init(sfd,EPOLLET|EPOLLIN,NULL,NULL);
    strcpy(epmsg->name,"signal");
    epmsg_add(logmor.epoll_fd,epmsg);
    // 创建 socket
    int server_socket;
    //if (fdrequest("",&server_socket,"logm",5)<0)
    server_socket=listen_unix(logmor.sock_path,SOCK_STREAM|SOCK_CLOEXEC,20);
    
    //epoll注册server_socket
    epmsg=epmsg_init(server_socket,EPOLLIN|EPOLLET|EPOLLHUP,NULL,NULL);
    strcpy(epmsg->name,"logm");
    epmsg_add(logmor.epoll_fd,epmsg);

    //epoll注册kmsg
    int kmsgfd=open("/dev/kmsg",O_CLOEXEC|O_RDONLY|O_NONBLOCK);
    if (kmsgfd>0)
    {
        epmsg=epmsg_init(kmsgfd,EPOLLIN|EPOLLET|EPOLLHUP,fdopen(kmsgfd,"r"),NULL);
        strcpy(epmsg->name,"kernel");
        epmsg_add(logmor.epoll_fd,epmsg);
    }
    
    // log持久化
    FILE *logpath=fopen(logmor.log_path,"a");

    //inotify
    // head_t info={
    //     .magic=CONN_MAGIC_INOTIFY,
    //     .action=CONN_INOTIFY_READY,
    //     .payloadlen=getpid()};
    // inotify("/run/rinitd/rinitd.sock",&info,sizeof(info));
    sigqueue(getppid(),SIGRTMIN+1,(union sigval)getpid());
    struct epoll_event evts[20];
    while (1)
    {
       // printf("新建连接client hhhhhhhhh fd:%d\n",client_fd);
        int ret=epoll_wait(logmor.epoll_fd,evts,20,-1);
        //printf("epoll ret:%d\n",ret);
        for (int i = 0; i < ret; i++)
        {
            epmsg_t *epmsg=(epmsg_t*)evts[i].data.ptr;
            if (epmsg->fd==server_socket) //socket触发连接事件
            {
                //socket新建连接
                int client_fd;
                if ((client_fd = accept(server_socket, NULL, NULL)) <= 0) {
                    if (logmor.kmsg!=NULL)
                    {
                        fprintf(logmor.kmsg,"logm[%d]: socket %s accept %d:\n",logmor.pid,strerror(errno),client_fd);
                        fflush(logmor.kmsg);
                    }
                }else{
                    int remote_fd;
                    char msg[1024];
                    if(recvfd(client_fd,&remote_fd,msg,1024)!=-1){
                        if (remote_fd==-1)
                        {
                            continue;
                        }
                        fdnonblock(remote_fd);
                        epmsg=epmsg_init(remote_fd,EPOLLIN|EPOLLET|EPOLLHUP,fdopen(remote_fd,"r"),NULL);
                        strcpy(epmsg->name,msg);
                        epmsg_add(logmor.epoll_fd,epmsg);
                        if (logmor.kmsg!=NULL)
                        {
                            fprintf(logmor.kmsg,"logm[%d]: add %s log fd=%d\n",logmor.pid,epmsg->name,remote_fd);
                            fflush(logmor.kmsg);
                        }
                    }
                }
                close(client_fd);
            }else if (epmsg->fd!=server_socket&&epmsg->fd!=sfd&&!(evts[i].events&EPOLLHUP))
            {
                freadlines(epmsg,epmsg->stream,logpath,4096,ioworker); //采集日志
            }else if (evts[i].events&EPOLLHUP)
            {
                fprintf(logmor.kmsg,"logm[%d]: del %s log fd=%d\n",logmor.pid,epmsg->name,epmsg->fd);
                fflush(logmor.kmsg);
                epmsg_del(logmor.epoll_fd,epmsg);
            }else if (epmsg->fd==sfd)
            {
                struct signalfd_siginfo fdsi;
                ssize_t s=read(epmsg->fd, &fdsi, sizeof(fdsi));
                if (s!=sizeof(fdsi))
                {
                    continue;
                }
                if (fdsi.ssi_signo==SIGINT || fdsi.ssi_signo==SIGTERM || fdsi.ssi_signo==SIGQUIT) //信号处理
                {
                    fprintf(logmor.kmsg,"logm[%d]: Exited Signal=%d\n",logmor.pid,fdsi.ssi_signo);  
                    //fprintf(logmor.kmsg,"logm[%d]: Exited Signal=\n",logmor.pid);  
                    //exit(EXIT_SUCCESS);
                    goto logm_exit;
                }
            }
            
        }
    }
logm_exit:
    // 关闭连接
    close(server_socket);
    fclose(logmor.kmsg);
    unlink(logmor.sock_path);
    return 0;
}

//日志轮转

// void pipe_handler(int sig){
//     char msgbuf[1024];
//     sprintf(msgbuf,".(%d)",getpid());
//     while (1)
//     {
//         printf("recv pipe signal retry connect log_server\n");
//         int local_fd=handle_fd(SOCKET_PATH,msgbuf);
//         if (local_fd>0)
//         {
//             dup2(local_fd,STDOUT_FILENO);
//             dup2(local_fd,STDERR_FILENO);
//             close(local_fd);
//             break;
//         }else{
//             usleep(30000000);
//         }
//     }
// }


// int client(char *cmd) 
// {
//     char msgbuf[1024];
//     sprintf(msgbuf,"[%s].\033[0;32m%d\033[0m",cmd,getpid());
//     int local_fd=handle_fd(SOCKET_PATH,msgbuf);
//     dup2(local_fd,STDOUT_FILENO);
//     dup2(local_fd,STDERR_FILENO);
//     close(local_fd);  
//     //信号处理
//     signal(SIGPIPE,pipe_handler);
//     //signal(SIGTTOU,stdout_handler);
//     system(cmd);
//     fflush(stdout);
//     fflush(stderr);
//     return 0;
// }